package com.yxc.consumerbath.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * @author: YXC
 * @time: 2022/12/7 11:16
 **/
@Slf4j
@Component
public class BathConsumer {

    public static final String TOPIC = "kafka-bigdata-demo-topic";

    @KafkaListener(topics = {TOPIC}, errorHandler = "myKafkaListenerErrorHandler", groupId = "${spring.kafka.consumer.group.demo-bigdata-topic-group}")
    public void consumer(List<ConsumerRecord<Object, Objects>> consumerRecordList) throws InterruptedException {
        //用于测试异常处理
        log.debug("收到消息: {}条", consumerRecordList.size());
        // 模拟处理100毫秒
        Thread.sleep(1000);
    }

}
